package com.testmq;

import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
 
import java.util.concurrent.*;
 
/**
 * @ClassName TransactionProducer
 * @Description TODO 事务消息机制
 *
 * 参考 https://github.com/apache/rocketmq/blob/master/docs/cn/RocketMQ_Example.md#6-%E6%B6%88%E6%81%AF%E4%BA%8B%E5%8A%A1%E6%A0%B7%E4%BE%8B
 */
public class TransactionProducer {
 
    public static void main(String[] args) throws Exception {
 
        // 这个东西就是用来接收RcoketMQ回调的一个监听器接口
        // 这里会实现执行订单本地事务，commit，rollback，回调查询等逻辑
        TransactionListener transactionListener = new TransactionListenerImpl();
 
        // 下面这个就是创建支持事务消息的Producer
        // 对这个Producer还得指定要指定一个生产者分组，根据业务指定名字
        TransactionMQProducer producer = new
                TransactionMQProducer("TestProducerGroup");
 
        producer.setNamesrvAddr("192.168.31.132:9876");
 
        // 下面这个是指定了一个线程池，里面会包含一些线程
        // 这个线程池里的线程就是用来处理RocketMQ回调你的请求
        ExecutorService executorService = new ThreadPoolExecutor(
                2,
                5,
                100,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
                    @Override
                    public Thread newThread(Runnable r) {
                        Thread thread = new Thread(r);
                        thread.setName("TestThread");
                        return thread;
                    }
        });
 
        // 给事务消息生产者设置对应的线程池，负责执行RocketMQ回调请求
        producer.setExecutorService(executorService);
        // 给事务消息生产者设置对应的回避函数
        producer.setTransactionListener(transactionListener);
        //启动实例
        producer.start();
 
 
        // 构造一条订单支付成功的消息，指定Topic是谁
        Message msg = new Message("PayOrderSuccessTopic" /* Topic */,
                "TestTag" /* Tag */,
                "TestKey",
                ("订单支付消息").getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
        );
        //Call send message to deliver message to one of brokers.
        SendResult sendResult = producer.sendMessageInTransaction(msg,null);
        System.out.printf("%s%n", sendResult);
    }
}